/*
 * Name     : Client.java
 * Author(s): Jill San Luis, Joshua Kwan
 * Created  : Apr 6, 2011, 12:44:06 AM
 */


import java.io.*;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.data.Stat;
import org.apache.zookeeper.KeeperException;

/**
 * The Client class
 *
 * @author  Jill San Luis
 * @author  Joshua Kwan
 */
public class Client {

    /** znode path specific for this class **/
    public static final String JOBTRACKER_PATH = "/jobtracker";

    /** manager for ZooKeeper server connection */
    private ZkConnector zkc;
	
    /** private ZooKeeper object */
    private ZooKeeper zk;
	
    /** watches the local zNode for responses from the jobtracker */
    private Watcher watcher;
    
    /** watches the jobtracker and reconnects upon failure */
    private Watcher jtWatcher;
    
    /** username */
    private String userID = "";
    
    /** path to client */
    private String clientPath; 
    
    /**
     * Creates a new instance of Client.
     */
    public Client(String host) {
    	zkc = new ZkConnector();
        try {
            zkc.connect(host);
        } catch(Exception e) {
            System.out.println("Zookeeper connect "+ e.getMessage());
        }
        
        zk = zkc.getZooKeeper();
        
        //create a watch on the /jobtracker node
    	jtWatcher = new Watcher(){
    		public void process(WatchedEvent event) {
        		handleJobTrackerEvent(event);
    		}
    	};
    	checkPath(JOBTRACKER_PATH, jtWatcher);
    }
    
    public static void main(String[] args){
    	if (args.length != 1) {
            System.out.println("Usage: java -classpath lib/zookeeper-3.3.2.jar:lib/log4j-1.2.15.jar:. A zkServer:clientPort");
            return;
        }
    	Client client = new Client(args[0]);
    	
    	BufferedReader stdIn = new BufferedReader(new InputStreamReader(System.in));
		String userInput;
		
		try {
			//get the userID from the client
			client.getUserID(stdIn);
			
			//create a watch on for the client node
			client.createClientWatcher();
						
			System.out.println("Enter queries or x for exit.");
			System.out.print("For new requests, enter \"new <hash>\", ");
			System.out.print("to check on previously submitted requests, enter \"query <hash>\":\n");
			System.out.print(">");
			while ((userInput = stdIn.readLine()) != null && !userInput.equals("x")){
				//check for valid userInput
				if (userInput.indexOf("new ") == 0 || userInput.indexOf("query ") == 0) {
					userInput = client.userID + " " + userInput;
					client.zkc.atomicAppend(userInput, Client.JOBTRACKER_PATH);
					//System.out.println("sent to JobTracker: " + userInput);
				} else {
					System.out.println("invalid entry");
				}
				System.out.print(">");
			}
		} catch (IOException e) {
			System.out.println(e);
		}
    }
    
    private void createClientWatcher() {
    	//create a watch on the /jobtracker node
    	watcher = new Watcher(){
    		public void process(WatchedEvent event) {
        		handleClientEvent(event);
    		}
    	};
    	checkPath(clientPath, watcher);
    }
    
    private void handleJobTrackerEvent(WatchedEvent event) {
    	String path = event.getPath();
        EventType type = event.getType();
        if(path.equalsIgnoreCase(JOBTRACKER_PATH)) {
            if (type == EventType.NodeDeleted) {
            	//try to reconnect until successful
                while (zkc.exists(JOBTRACKER_PATH, jtWatcher) == null);
            }
        } else {
        	checkPath(JOBTRACKER_PATH, jtWatcher);
        }
    }
    
    private void handleClientEvent(WatchedEvent event) {
    	String path = event.getPath();
        EventType type = event.getType();
        if(path.equalsIgnoreCase(clientPath)) {
            if (type == EventType.NodeDataChanged) {
//           	try {
/*            		Stat stat = new Stat();
            		byte[] clientData = zk.getData(clientPath, watcher, stat);
            		if (clientData != null || new String(clientData).equalsIgnoreCase("")){
                    	String strClientData = new String(clientData);
*/					String strClientData = zkc.atomicGetAndClearData(clientPath);
					checkPath(clientPath, watcher);
					if (!strClientData.equalsIgnoreCase("")){
            			String [] clientDataArray = strClientData.split(ZkConnector.DATADELIMITER);
                    	for (int i = 0; i < clientDataArray.length; i++){
                    		if (!clientDataArray[i].equalsIgnoreCase("")){
                    			System.out.print(clientDataArray[i] + "\n");
                        		System.out.print(">");	
                    		}
                    	}            			
            		}
/*            	} catch (KeeperException e) {
            		System.out.println(e);
            	} catch (InterruptedException e) {
            		System.out.println(e);
            	}
*/            } else {
            	checkPath(clientPath, watcher);
            }
            if (watcher == null) System.out.println("Watcher for " + clientPath + " is null!!");
        }
    }

    private Stat checkPath(String path, Watcher watcher) {
        Stat stat = zkc.exists(path, watcher);
        return stat;
    }
    
    private void getUserID (BufferedReader stdIn) throws IOException{
    	while (userID.equals("")){
			System.out.print("Enter a unique userID:\n");
			System.out.print(">");
			userID = stdIn.readLine();
			if (!userID.equals("")){
				//try to create a new ephemeral node under /userID
				//proceed if successful
				//set client.userID to "" if node creation fails 
				clientPath = "/" + userID;
				Stat stat = zkc.exists(clientPath, false);
				if (stat == null) {              // znode doesn't exist; let's try creating it
		            System.out.println("Creating " + clientPath);
		            Code ret = zkc.create(
		            			clientPath,         // Path of znode
		                        null,           // Data not needed.
		                        CreateMode.EPHEMERAL   // Znode type, set to EPHEMERAL.
		                        );
		            if (ret != Code.OK){
		            	userID = "";
		            }
		        } else {
		        	userID = "";
		        }
			}
		}
    }
    
    
}
